package jk.fox.exchange.client;

import jk.fox.common.RpcDict;
import jk.fox.common.RpcStatus;
import jk.fox.common.RpcURIUtils;
import jk.fox.common.exception.runtime.RpcConnectRetryException;
import jk.fox.common.payload.request.RpcRequest;
import jk.fox.common.payload.response.Response;
import jk.fox.config.exchange.standard.DefaultExchangeConfiguration;
import jk.fox.exchange.ExchangeChannel;
import jk.fox.node.NodeStatusListener;
import jk.fox.transport.ClientTransporter;
import org.apache.log4j.Logger;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @author cuichao
 * @Description: NettyClientExchange
 * @Date: create in 2020/3/28 21:37
 */
public class DefaultClientExchange implements ClientExchange<ExchangeChannel> {

    private static Logger logger = Logger.getLogger(DefaultClientExchange.class);

    private ClientTransporter transporter;

    private List<NodeStatusListener> listeners;

    private volatile ConcurrentMap<String,ExchangeChannel> channelPool;

    private int connectMaxRetries;

    private long ConnectRetryIntervalMS;

    private Lock lock = new ReentrantLock();


    public DefaultClientExchange(DefaultExchangeConfiguration config, ClientTransporter transporter){
        listeners = new ArrayList<>();
        channelPool = new ConcurrentHashMap<>(32);
        connectMaxRetries = config.getConnectMaxRetries();
        ConnectRetryIntervalMS = config.getConnectRetryIntervalMS();
        this.transporter = transporter;
    }

    @Override
    public ExchangeChannel connection(InetSocketAddress address) throws Exception {
        String key = RpcURIUtils.getAddress(address);
        ExchangeChannel channel = channelPool.get(key);
        if(!isActive(channel)){
            lock.lock();
            try {
                channel = channelPool.get(key);
                int tryCounter = 0;
                while (!isActive(channel) && tryCounter++ < connectMaxRetries){
                    try {
                        channel = transporter.doConnection(address);
                    }catch (Exception e){
                        TimeUnit.MILLISECONDS.sleep(ConnectRetryIntervalMS);
                    }
                    if(!isActive(channel)){
                        TimeUnit.MILLISECONDS.sleep(ConnectRetryIntervalMS);
                    }
                }
                if(!isActive(channel)){
                    RpcConnectRetryException e = new RpcConnectRetryException(String.format("the serverTransporter has max connectMaxRetries connection: %d", connectMaxRetries));
                    logger.error(String.format("the serverTransporter has max connectMaxRetries connection: %d", connectMaxRetries));
                    //连接失败触发服务端状态监听
                    publishServerStatus(key,RpcDict.NodeStatus.FAILURE,e);
                    logger.debug("trigger server status listener. the status is "+RpcDict.NodeStatus.FAILURE,e);
                    throw e;
                }
                //连接成功触发服务端状态监听
                logger.debug("trigger server status listener. the status is "+RpcDict.NodeStatus.SUCCESS);
                publishServerStatus(key,RpcDict.NodeStatus.SUCCESS,null);
                channelPool.put(key, channel);
            }finally {
                lock.unlock();
            }
        }
        return channel;
    }

    @Override
    public Object execute(URI uri, Object[] parameters, String[] parameterTypes) throws Exception {
        return execute(uri,parameters,parameterTypes,0);
    }

    @Override
    public Object execute(URI uri, Object[] parameters, String[] parameterTypes, long timeout) throws Exception {
        InetSocketAddress socketAddress = RpcURIUtils.getInetSocketAddress(uri);
        ExchangeChannel channel = connection(socketAddress);
        String path = RpcURIUtils.getPath(uri);
        RpcRequest request = new RpcRequest();
        request.setRequestId(UUID.randomUUID().toString().replaceAll("-",""));
        request.setParameterTypes(parameterTypes);
        request.setPath(path);
        request.setParameters(parameters);
        Response response = transporter.send(channel, request,timeout);
        //返回异常判断
        if(RpcStatus.isOk(response.getResponseCode())){
            return response.getResponse();
        }else {
            throw RpcStatus.getBadRequestException(response.getResponseCode(),response.getExceptionMessage());
        }

    }

    @Override
    public boolean isActive(ExchangeChannel channel) {
        return channel != null && channel.isAlive();
    }

    @Override
    public void addServerStatusListener(NodeStatusListener listener) {
        listeners.add(listener);
    }

    @Override
    public void publishServerStatus(String address, RpcDict.NodeStatus status, Exception e) {
        for (NodeStatusListener listener : listeners) {
            listener.listen(address,status,e);
        }
    }




}

